package studio.raptor.dispatcher.server.task;

import java.io.IOException;
import java.util.concurrent.locks.LockSupport;
import javax.annotation.PreDestroy;
import javax.jms.JMSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusConsumer;
import studio.raptor.databus.client.DatabusConsumerBuilder;
import studio.raptor.databus.client.EventMessage;
import studio.raptor.databus.command.Event;
import studio.raptor.dispatcher.server.context.HandlerContext;
import studio.raptor.dispatcher.server.dto.Consumer;
import studio.raptor.dispatcher.server.handler.consumption.FilterHandler;
import studio.raptor.dispatcher.server.handler.consumption.IdempotenceHandler;
import studio.raptor.dispatcher.server.handler.consumption.PersistEventHandler;
import studio.raptor.dispatcher.server.handler.consumption.SortHandler;
import studio.raptor.dispatcher.server.properties.MQProperties;
import studio.raptor.gid.Sequencer;

/**
 * @author jack
 * @since 1.0
 */
@Component
@Scope("prototype")
public class ConsumerTask implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(ConsumerTask.class);

  private Consumer consumer;

  private volatile boolean start;

  @Autowired
  FilterHandler filterHandler;
  @Autowired
  IdempotenceHandler idempotenceHandler;
  @Autowired
  PersistEventHandler persistEventHandler;
  @Autowired
  SortHandler sortHandler;
  @Autowired
  MQProperties mqProperties;
  @Autowired
  Sequencer sequencer;

  DatabusConsumer databusConsumer;

  @PreDestroy
  public void destory() {
    log.info("the consumer:{}-{},close mq client", consumer.getCode(), consumer.getName());
    databusConsumer.close();
  }


  @Override
  public void run() {
    log.info("the consumer:{},kindId:{},state:{}", consumer.getCode(), consumer.getKindId(),
        start ? "start" : "stop");
    initProcess();
    initMq();

    try {
      while (!Thread.currentThread().isInterrupted()) {
        if (isStart()) {
          EventMessage message = null;
          try {
            message = databusConsumer.receiveUnACKMessage(mqProperties.getReceiveTimeout());
            if (null == message) {
              log.info("consumer [{}-{}] get no message,sleep 500 ms", consumer.getCode(),
                  consumer.getKindId());
              continue;
            }
            Event event = message.getEvent();

            HandlerContext context = buildContext(event);
            boolean result = filterHandler.process(context);
            log.info("consumer [{}] spending event [{}]result:{}", consumer.getCode(),
                context.getEvent().getEventMetaData().getBusinessNo(), result);
          } catch (Exception e) {
            log.error("the consumer:{},kindId:{}, exception", consumer.getCode(),
                consumer.getKindId(), e);
          } finally {
            if (message != null) {
              try {
                message.acknowledge();
              } catch (JMSException e) {
                e.printStackTrace();
              }
            }
          }
        } else {
          //任务暂停，休眠500ms,防止cpu过高
          log.info("suspend consumer [{}-{}] , sleep for 5 seconds", consumer.getCode(),
              consumer.getKindId());
          LockSupport.parkUntil(System.currentTimeMillis() + 500);
        }
      }
    } catch (Exception e) {
      log.error("the consumer:{},kindId:{}, exception", consumer.getCode(), consumer.getKindId(),
          e);
    }

  }

  public Consumer getConsumer() {
    return consumer;
  }

  public void setConsumer(Consumer consumer) {
    this.consumer = consumer;
  }

  public boolean isStart() {
    return start;
  }

  public synchronized void setStart(boolean start) {
    this.start = start;
  }

  private DatabusConsumer createConsumer() throws IOException, JMSException {
    DatabusConfig databusConfig = new DatabusConfig(mqProperties.getBizKey());
    DatabusConsumerBuilder databusConsumerBuilder = new DatabusConsumerBuilder(databusConfig);
    return databusConsumerBuilder
        .create(consumer.getEventName(), consumer.getKindId());
  }

  private HandlerContext buildContext(Event event) throws Exception {
    HandlerContext context = new HandlerContext();
    context.setName(consumer.getName());
    context.setCode(consumer.getCode());
    context.setKindId(consumer.getKindId());
    context.setEventType(consumer.getEventType());
    context.setTraceId(sequencer.get("traceId").nextId());
    context.setEvent(event);
    context.setSort(consumer.isSort());
    return context;
  }

  private void initProcess() {
    boolean sort = consumer.isSort();
    filterHandler.setSuccessor(idempotenceHandler);
    idempotenceHandler.setSuccessor(persistEventHandler);

    if (sort) {
      persistEventHandler.setSuccessor(sortHandler);
    }
  }

  private void initMq() {
    try {
      databusConsumer = createConsumer();
    } catch (IOException e) {
      e.printStackTrace();
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }

}
